﻿using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Mrc.Config;
using Mrc.Config.Messages;
using Mrc.RabbitMQ;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Mrc.RabbitMQ
{
    public class RabbitmqConsumer : BackgroundService
    {
        private readonly ILogger _logger;
        private readonly ISerializer _serializer;
        private IModel _channel;
        private MrcConfigOptions _mrcConfigOptions;
        private readonly IConnectionPool _connectionPool;
        private string _exchange;
        private string _queue;
        private string _routingKey;
        private readonly INodeHealthCheckService _nodeHealthCheckService;
        public RabbitmqConsumer(ILoggerFactory loggerFactory,
            IOptions<MrcConfigOptions> mrcConfigOptions,
            ISerializer serializer,
            IConnectionPool connectionPool,
            INodeHealthCheckService nodeHealthCheckService
            )
        {
            this._nodeHealthCheckService = nodeHealthCheckService;
            this._connectionPool = connectionPool;
            this._serializer = serializer;
            this._logger = loggerFactory.CreateLogger<RabbitmqConsumer>();
            _mrcConfigOptions = mrcConfigOptions.Value;

            _exchange = $"{_connectionPool.Exchange}.{_mrcConfigOptions.ItemName}";
            _queue = $"mrc.queue.{_mrcConfigOptions.ItemName}.log";
            _routingKey = $"mrc.queue.{_mrcConfigOptions.ItemName}.*";
            StartConsumer();
        }

        private void StartConsumer()
        {
            _channel = _connectionPool.Rent();
            _channel.ExchangeDeclare(_exchange, ExchangeType.Fanout);
            _channel.QueueDeclare(_queue, false, false, false, null);
            _channel.QueueBind(_queue, _exchange, _routingKey, null);
            _channel.BasicQos(0, 1, false);
        }

        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            stoppingToken.ThrowIfCancellationRequested();
            var consumer = new EventingBasicConsumer(_channel);
            consumer.Received += (ch, ea) =>
            {
                HandleMessage(Encoding.UTF8.GetString(ea.Body.Span));
                _channel.BasicAck(ea.DeliveryTag, false);
            };
            consumer.Shutdown += OnConsumerShutdown;
            consumer.Registered += OnConsumerRegistered;
            consumer.Unregistered += OnConsumerUnregistered;
            consumer.ConsumerCancelled += OnConsumerConsumerCancelled;
            _channel.BasicConsume($"mrc.queue.{_mrcConfigOptions.ItemName}.log", false, consumer);
            return Task.CompletedTask;
        }

        private void HandleMessage(string content)
        {
            _logger.LogInformation($"consumer received {content}");
            var result = _serializer.DeserializeAsync(content).GetAwaiter().GetResult();
            switch (result.MsgType)
            {
                case NoticeMsgTypeEnum.ConfigMsg:
                    LocationConfigStore.Set(result.Values);
                    break;
                case NoticeMsgTypeEnum.HealthCheckMsg:
                    _nodeHealthCheckService.UpHealthInfo();
                    break;
                default:
                    break;
            }
        }
        private void OnConsumerConsumerCancelled(object sender, ConsumerEventArgs e) { }
        private void OnConsumerUnregistered(object sender, ConsumerEventArgs e) { }
        private void OnConsumerRegistered(object sender, ConsumerEventArgs e) { }
        private void OnConsumerShutdown(object sender, ShutdownEventArgs e) { }
        private void RabbitMQ_ConnectionShutdown(object sender, ShutdownEventArgs e) { }

        public override void Dispose()
        {
            _channel.Close();
            base.Dispose();
        }
    }
}